package MountainMq.service.impl;


import MountainMq.dao.TransMessageDao;
import MountainMq.enummeration.TransMessageType;
import MountainMq.po.TransMessagePO;
import MountainMq.service.TransMessageService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.UUID;

@Service
public class TransMessageServiceImpl implements TransMessageService {
    @Autowired
    private TransMessageDao transMessageDao;
    @Value("${mountainmq.service}")
    private String serviceName;

    /**
     * 发送前暂存消息
     *
     * @param exchange exchange
     * @param routingKey routingKey
     * @param body body
     * @return TransMessagePO
     */
    @Override
    public TransMessagePO messageSendReady(String exchange, String routingKey, String body) {
        final String messageId = UUID.randomUUID().toString();
        // 构建实体类
        TransMessagePO transMessagePO = new TransMessagePO();
        transMessagePO.setId(messageId);
        transMessagePO.setService(serviceName);
        transMessagePO.setExchange(exchange);
        transMessagePO.setRoutingKey(routingKey);
        transMessagePO.setPayload(body);
        transMessagePO.setSequence(0);
        transMessagePO.setType(TransMessageType.SEND);
        transMessagePO.setDate(new Date());
        transMessageDao.insert(transMessagePO);
        return transMessagePO;
    }

    /**
     * 设置消息发送成功，需要把消息删掉
     * @param id 消息ID
     */
    @Override
    public void messageSendSuccess(String id) {
        transMessageDao.delete(id,serviceName);
    }

    /**
     * 设置消息返回 消息重新持久化
     * @param messageId 消息ID
     * @param exchange  exchange
     * @param routingKey routingKey
     * @param body body
     * @return TransMessagePO
     */
    @Override
    public TransMessagePO messageSendReturn(String messageId, String exchange, String routingKey, String body) {
        // 根据之前的消息ID又重新持久化 这个目前有问题 插入是很有可能报错的，如果使用的是逻辑删除
//        TransMessagePO transMessagePO = new TransMessagePO();
//        transMessagePO.setId(messageId);
//        transMessagePO.setService(serviceName);
//        transMessagePO.setExchange(exchange);
//        transMessagePO.setRoutingKey(routingKey);
//        transMessagePO.setPayload(body);
//        transMessagePO.setSequence(0);
//        transMessagePO.setType(TransMessageType.SEND);
//        transMessagePO.setDate(new Date());
//        transMessageDao.insert(transMessagePO);
//        return transMessagePO;

        // 消息持久化就是这个方法 原代码 但是根本没用到messageId，也就是说相当于重新创建了一条数据
        return messageSendReady(exchange, routingKey, body);
    }

    /**
     * 查询应发未发消息
     * @return List<TransMessagePO>
     */
    @Override
    public List<TransMessagePO> listReadyMessages() {
        return transMessageDao.selectByTypeAndService(TransMessageType.SEND.toString(),serviceName);
    }

    /**
     * 记录(增加)消息发送次数
     * @param id id
     */
    @Override
    @Transactional // 使用行锁，但是最好方法是使用分布式锁，以后有时间再改
    public void messageResend(String id) {
        TransMessagePO transMessagePO = transMessageDao.selectByIdAndService(id, serviceName);
        // 这里可能有并发问题
        transMessagePO.setSequence(transMessagePO.getSequence()+1);
        transMessageDao.update(transMessagePO);
    }

    /**
     * 消息重发多次，放弃
     * @param id id
     */
    @Override
    public void messageDead(String id) {
        TransMessagePO transMessagePO = transMessageDao.selectByIdAndService(id, serviceName);
        transMessagePO.setType(TransMessageType.DEAD);
        transMessageDao.update(transMessagePO);
    }

    /**
     * 保存监听到的死信消息
     * @param id
     * @param exchange
     * @param routingKey
     * @param queue
     * @param body
     */
    @Override
    public void messageDead(String id, String exchange, String routingKey, String queue, String body) {
        TransMessagePO transMessagePO = new TransMessagePO();
        transMessagePO.setId(id);
        transMessagePO.setService(serviceName);
        transMessagePO.setExchange(exchange);
        transMessagePO.setRoutingKey(routingKey);
        transMessagePO.setPayload(body);
        transMessagePO.setDate(new Date());
        transMessagePO.setSequence(0);
        transMessagePO.setType(TransMessageType.DEAD);
        transMessageDao.insert(transMessagePO);
    }

    /**
     * 消息消费前保存
     * @param id
     * @param exchange
     * @param routingKey
     * @param queue
     * @param body
     * @return
     */
    @Override
    public TransMessagePO messageReceiveReady(String id, String exchange, String routingKey, String queue, String body) {
        // 查询消息的该消息是否存在于数据库
        TransMessagePO transMessagePO = transMessageDao.selectByIdAndService(id,serviceName);
        if(null == transMessagePO){
            transMessagePO = new TransMessagePO();
            transMessagePO.setId(id);
            transMessagePO.setService(serviceName);
            transMessagePO.setExchange(exchange);
            transMessagePO.setRoutingKey(routingKey);
            transMessagePO.setQueue(queue);
            transMessagePO.setPayload(body);
            transMessagePO.setDate(new Date());
            transMessagePO.setSequence(0);
            // 修改状态为receive
            transMessagePO.setType(TransMessageType.RECEIVE);
            transMessageDao.insert(transMessagePO);
        }else{ // 消息存在
            // 增加次数就行了  不存在并发问题，因为MQ每次只会把消息推给一个客户端
            transMessagePO.setSequence(transMessagePO.getSequence()+1);
            transMessageDao.update(transMessagePO);
        }
        return transMessagePO;
    }

    /**
     * 消息消费成功
     * @param id
     */
    @Override
    public void messageReceiveSuccess(String id) {
        transMessageDao.delete(id,serviceName);
    }
}
